package com.yjd.msg.comm.listener;

import com.alibaba.fastjson.JSON;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.msg.MnsMsgInfo;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.SpringContextHolder;
import com.yjd.comm.util.StringUtil;
import com.yjd.msg.comm.enums.MsgStatusEnum;
import com.yjd.msg.comm.model.MsgDisposeLogModel;
import com.yjd.msg.comm.model.MsgDisposeResult;
import com.yjd.msg.comm.model.MsgMappedDisposeModel;
import com.yjd.msg.comm.service.IMsgDisposeLogService;
import com.yjd.msg.comm.service.IMsgInfoService;
import com.yjd.msg.comm.service.IMsgMappedDisposeService;
import com.yjd.msg.comm.util.MnsUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.BeanUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * Created by yjd-wap on 2018/1/27.
 */
public class MsgDisposeUtil {

    private static Logger logger = Logger.getLogger(MsgDisposeUtil.class);

    private static IMsgMappedDisposeService getMsgMappedDisposeService()throws Exception{
        return   SpringContextHolder.getBean("msgMappedDisposeService");
    }

    private static IMsgDisposeLogService getMsgDisposeLogService()throws Exception{
        return   SpringContextHolder.getBean("msgDisposeLogService");
    }

    private static IMsgInfoService getMsgInfoService()throws Exception{
        return   SpringContextHolder.getBean("msgInfoService");
    }

    private static ExecutorService threadpool = Executors.newCachedThreadPool();

    /**
     * 消费消息
     * @param mnsMsgInfo
     */
    public static void onMessage(MnsMsgInfo mnsMsgInfo, String receiptHandle, String queueName) {
        logger.info("开始消费消息："+ JSON.toJSONString(mnsMsgInfo)+" receiptHandle:"+receiptHandle+ "queueName:"+queueName);
        MsgStatusEnum status = MsgStatusEnum.CONSUME_SUCCESS;
        Long currtime = FrameUtil.getTime(null);
        try {
            List<MsgMappedDisposeModel> messageListeners= getMsgMappedDisposeService().getModelList(
                    FrameUtil.newHashMap("msg_type",mnsMsgInfo.getMsg_type()), DbWREnums.WRITE);
            if (messageListeners.isEmpty()) {
                logger.error("没有找到消息的消费者:msg_id" + mnsMsgInfo.getMsg_id() + ",消息类型" + mnsMsgInfo.getMsg_type());
                return;
            }

            int listerSize = messageListeners.size();
            CountDownLatch countDownLatch = new CountDownLatch(listerSize);
            CompletionService<MsgDisposeResult> completionService = new ExecutorCompletionService<>(threadpool);
            MnsMsgInfo threadMessage = null;
            List<MsgDisposeLogModel> disposeLogModels = new ArrayList<MsgDisposeLogModel>();
            for (MsgMappedDisposeModel messageListener : messageListeners) {
                // 创建message的副本,确保线程安全
                threadMessage = new MnsMsgInfo();
                BeanUtils.copyProperties(mnsMsgInfo, threadMessage);
                // 多线程调用
                logger.info("调用监听器服务名称:" + messageListener.getHandle_service() );
                completionService.submit(new MnsMsgConsumeTask(messageListener.getHandle_service(), threadMessage, countDownLatch));
            }
            // 同步
            countDownLatch.await();


            //判断线程池处理是否全部成功，并且记录每个线程处理的结果
            for (int i = 0;i<listerSize;i++) {
                MsgDisposeResult disposeResult = completionService.take().get();
                if(disposeResult.getMsgStatusEnum()==null||disposeResult.getMsgStatusEnum().getCode()==MsgStatusEnum.CONSUME_FAIL.getCode()){
                    status = MsgStatusEnum.CONSUME_FAIL;
                }
                //记录多个处理器执行的结果
                disposeLogModels.add(new MsgDisposeLogModel(mnsMsgInfo.getMsg_id(),disposeResult.getListerServiceName(),mnsMsgInfo.getMsg_type(),
                        disposeResult.getMsgStatusEnum()==null?MsgStatusEnum.CONSUME_FAIL.getCode():disposeResult.getMsgStatusEnum().getCode(),currtime));
            }


            //记录处理结果
            getMsgDisposeLogService().batchInsertModel(disposeLogModels);
            if((MsgStatusEnum.CONSUME_SUCCESS.getCode()==status.getCode())&&(!StringUtil.isEmpty(receiptHandle))&&(!StringUtil.isEmpty(queueName))){
                //全部执行成功则删除阿里云消息
                MnsUtils.getAliyunClient().getQueueRef(queueName).deleteMessage(receiptHandle);
            }
            //更新消息记录状态
            getMsgInfoService().finished(threadMessage,status);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}
